/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.nhncorp.neptune.tabletserver.action;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.nhncorp.neptune.common.MutableInteger;
import com.nhncorp.neptune.common.NeptuneLock;
import com.nhncorp.neptune.tablet.TabletInfo;

public class ActionChecker {
  public static final Log LOG = LogFactory.getLog(ActionChecker.class.getName());

  protected Actions runningActions = new Actions();
  
  protected Actions waitingActions = new Actions();
  
  protected final NeptuneLock lock = new NeptuneLock();
  
  protected ActionChecker actionChecker;
  
  protected TabletInfo tabletInfo;
  
  protected ThreadGroup threadGroup;
  
  private ExecutorService compactionExecutor;
  
  private ExecutorService splitExecutor;

  private ExecutorService actionExecutor;

  //FIXME Sync에 대해 정리 필요, 가끔 MajorCompaction과 MinorCompaction이 같이 수행되는 경우가 발생하여
  //현재는 모두 sync 처리 했음 -> 성능 확인 및 적절하게 lock 구성 필요
  public ActionChecker(
      ExecutorService compactionExecutor,
      ExecutorService splitExecutor,
      ExecutorService actionExecutor,
      ThreadGroup threadGroup, TabletInfo tabletInfo) {
    this.actionChecker = this;
    this.threadGroup = threadGroup;
    this.tabletInfo = tabletInfo;
    this.splitExecutor = splitExecutor;
    this.compactionExecutor = compactionExecutor;
    this.actionExecutor = actionExecutor;
  }
  
  public Collection<String> getRunningActions() {
    lock.obtainReadLock();
    try {
      return runningActions.getListForReport();
    } finally {
      lock.releaseReadLock();
    }
  }
  
  public Collection<String>  getWaitingActions() {
    lock.obtainReadLock();
    try {
      return waitingActions.getListForReport();
    } finally {
      lock.releaseReadLock();
    }      
  }
  
  private boolean startAction(ActionIF action, boolean addWaitingList) {
    lock.obtainWriteLock();
    try {
      boolean canRun = action.canRun(this);
      if(canRun) {
        if(action.isThreadMode()) { 
          startActionThread(action);
        }
        
        //LOG.debug("Start action:" + action.getActionKey());
        runningActions.add(action);
      } else {
        if(addWaitingList && action.isWaitingMode() && !hasAlreadyScheduled(action)) {
          waitingActions.add(action);
        } else {
        }
      }
      return canRun;
    } finally {
      lock.releaseWriteLock();
    }
  }
  
  public boolean hasAlreadyScheduled(ActionIF action) {
    return runningActions.contains(action.getActionType()) 
      || waitingActions.contains(action.getActionType());
  }
  
  public boolean startAction(ActionIF action) {
    return startAction(action, true);
  }
  
  public void endAction(TabletAction action) {
    lock.obtainWriteLock();
    try {
      runningActions.remove(action);
//      try {
//        if(tabletActionListener != null) {
//          tabletActionListener.endAction(action);
//        }
//      } catch (Exception e) {
//        LOG.error("startActionThread:" + e.getMessage());
//        e.printStackTrace(System.out);
//      }
    } finally {
      lock.releaseWriteLock();
    }
    waitingActions.startWaitAction();
  }
  
  public void endActionType(TabletAction action) {
    lock.obtainWriteLock();
    try {
      runningActions.removeType(action); 
      waitingActions.removeType(action);
    } finally {
      lock.releaseWriteLock();
    }
    //LOG.debug("endActionType:" + action.getActionType() + ", tablet=" + tabletInfo.getTabletName());
    waitingActions.startWaitAction();
  }
  
  private void startActionThread(ActionIF action){
    if(MajorCompactionAction.ACTION_TYPE.equals(action.getActionType())) {
      compactionExecutor.execute(action);
    } else if(TabletSplitAction.ACTION_TYPE.equals(action.getActionType())) {
      splitExecutor.execute(action);
    } else {
      actionExecutor.execute(action);
    }
  }
  
  public void endActions() {
    lock.obtainWriteLock();
    try {
      runningActions.clear();
      waitingActions.clear();
    } finally {
      lock.releaseWriteLock();
    }      
  }

  public boolean hasAction(String actionKey) {
    lock.obtainWriteLock();
    try {
      return runningActions.actionMap.containsKey(actionKey) || 
                waitingActions.actionMap.containsKey(actionKey);
    } finally {
      lock.releaseWriteLock();
    }
  }
  
  public void print() {
    LOG.debug("=====Running Actions=====");
    runningActions.print();
    LOG.debug("=====Waiting Actions=====");
    waitingActions.print();
  }
  
  public List<String> getAllActions() {
    List<String> actionNames = new ArrayList<String>();
    actionNames.add("[Running Actions]");
    actionNames.addAll(runningActions.getListForReport());
    actionNames.add("[Waiting Actions]");
    actionNames.addAll(waitingActions.getListForReport());
    return actionNames;
  }  
  
  class Actions {
    //action key
    List<String> actionList = new ArrayList<String>();
    
    //action key
    Map<String, ActionInfo> actionMap = new HashMap<String, ActionInfo>();
    
    //action type
    Map<String, MutableInteger> actionTypeList = new HashMap<String, MutableInteger>();
    
    public void add(ActionIF action) {
        actionList.add(action.getActionKey());
        actionMap.put(action.getActionKey(), new ActionInfo(action));
        
        MutableInteger count = actionTypeList.get(action.getActionType());
        if(count == null) {
          count = new MutableInteger(0);
          actionTypeList.put(action.getActionType(), count);
        }
        count.add();
    }
    
    public int getActionCount(String actionKey) {
        MutableInteger count = actionTypeList.get(actionKey);
        if(count == null)   return 0;
        return count.getValue();
    }
    
    public void print() {
        String tabletName = "";
        if(tabletInfo != null)    tabletName = tabletInfo.getTabletName();
        for(String eachAction: actionList) {
          ActionInfo actionInfo = actionMap.get(eachAction);
          LOG.debug(tabletName + ":" + actionInfo.toString());
        }
    }

    public Collection<String> getListForReport() {
      List<String> result = new ArrayList<String>();
      for(String eachAction: actionList) {
        ActionInfo actionInfo = actionMap.get(eachAction);
        if(actionInfo != null) {
          result.add(actionInfo.getTabletAction().getActionKey() + ":" + new Date(actionInfo.getTime()));
        }
      }
      return result;
    }

    public synchronized void startWaitAction() {
      List<String> tempActionList = new ArrayList<String>();
      tempActionList.addAll(actionList);
      
      for(String eachActionKey: tempActionList) {
        ActionInfo actionInfo = actionMap.get(eachActionKey);
        ActionIF action = actionInfo.getTabletAction();
        if(action.canRun(actionChecker) && action.isThreadMode()) {
          if(startAction(action, false)) {
            remove(action);
          }
        }
      }
    }
    
    public synchronized void clear() {
      actionList.clear();
      actionMap.clear();
      actionTypeList.clear();
    }
    
    public synchronized void remove(ActionIF action) {
      String actionKey = action.getActionKey();
      actionList.remove(actionKey);
      actionMap.remove(actionKey);
      
      MutableInteger count = actionTypeList.get(action.getActionType());
      if(count != null) {
        count.minus();
        if(count.getValue() == 0) {
          actionTypeList.remove(action.getActionType());
        }
      }
    }

    public synchronized void removeType(ActionIF action) {
      //String actionKey = action.getActionKey();
      List<String> removeActionIds = new ArrayList<String>();
      for(Map.Entry<String, ActionInfo> entry: actionMap.entrySet()) {
        ActionInfo actionInfo = entry.getValue();
        if(actionInfo.getTabletAction().getClass().getName().equals(action.getClass().getName())) {
          removeActionIds.add(entry.getKey());
        }
      }
      
      for(String eachActionId: removeActionIds) {
        actionList.remove(eachActionId);
        actionMap.remove(eachActionId);
      }
      
      MutableInteger count = actionTypeList.get(action.getActionType());
      if(count != null) {
        actionTypeList.remove(action.getActionType());
      }
    }
    
    public boolean contains(String actionType) {
      return actionTypeList.containsKey(actionType);
    }
  }
  
  class ActionInfo {
    ActionIF tabletAction;
    long time;
    
    public ActionInfo(ActionIF tabletAction) {
      this.tabletAction = tabletAction;
      this.time = System.currentTimeMillis();
    }

    public ActionIF getTabletAction() {
      return tabletAction;
    }

    public long getTime() {
      return time;
    }
    
    public String toString() {
      return tabletAction.getActionKey() + ":" + new Date(time);
    }
  }
  
  public static void main(String[] args) {
    ActionChecker checker = new ActionChecker(
        Executors.newFixedThreadPool(5),
        Executors.newFixedThreadPool(5),
        Executors.newFixedThreadPool(5),
        new ThreadGroup("test"), null);
    checker.startAction(new BatchUploadAction());
    checker.startAction(new BatchUploadAction());
    checker.startAction(new BatchUploadAction());
    checker.startAction(new MajorCompactionAction(null));
    for(String eachAction: checker.getAllActions()) {
      System.out.println(eachAction);
    }
    
    //checker.print();
    checker.endActionType(new BatchUploadAction());
    //checker.print();
    for(String eachAction: checker.getAllActions()) {
      System.out.println(eachAction);
    }
  }
}
